package com.operator

import org.apache.spark.{SparkConf, SparkContext}

object FilterCounttoPerator {
  def main(args: Array[String]): Unit = {
//    System.setProperty("hadoop.home.dir", "E:\\hadoop")
    val sparkConf = new SparkConf().setAppName("MapOperator")
//      .setMaster("local")
    val sc = new SparkContext(sparkConf)

    val rdd1 = sc.textFile("hdfs://zhangzeli-node1:9000/input/log.txt")
    val rdd2 = rdd1.map(_.split(" ")(1))
    val rdd3 = rdd2.map((_,1))

    val rdd4 =rdd3.reduceByKey(_+_)

    /**
      * 交换k-v位置
      */
    val rdd5 = rdd4.map(_.swap)

    /**
      * 降序排序
      */
    val rdd6 = rdd5.sortByKey(false)

    val arr = rdd6.take(1)
    val schoolName = arr(0)._2


    /**
      * 出现次数最多的这个学校名字有了，对rdd1进行锅炉，在过滤的时候得一句shcoolName
      */
    val resultRDD = rdd1.filter(x=>{
      !schoolName.equals(x.split(" ")(1))
    })

    resultRDD.saveAsTextFile("hdfs://zhangzeli-node1:9000/output/result")

    sc.stop()
  }
}
